package com.ysw.rpc.client.utils;

import com.ysw.rpc.client.component.ChannelComponent;
import com.ysw.rpc.common.messages.Request;
import com.ysw.rpc.common.messages.Response;
import io.netty.channel.Channel;
import io.netty.util.concurrent.DefaultEventExecutor;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.nio.channels.ClosedChannelException;
import java.util.Objects;

import static com.ysw.rpc.client.utils.CacheUtil.getClassServerCache;

/**
 * @author YinShangwen
 * @since 2023/4/10 9:52
 */
@Slf4j
public class ChannelUtil {
    private static ChannelComponent channelComponent;

    public static void setChannelComponent(ChannelComponent channelComponent) {
        ChannelUtil.channelComponent = channelComponent;
    }

    /**
     * 发消息
     *
     * @param request 请求对象
     * @param clazz   接口class
     * @return promise
     */
    public static Promise<Response> sendMsg(Request request, Class<?> clazz) {
        DefaultEventExecutor executor = new DefaultEventExecutor(channelComponent.getEventLoopGroup());
        Promise<Response> promise = new DefaultPromise<>(executor);
        PromiseUtil.put(request, promise);

        sendMsg0(request, clazz);

        promise.addListener(future -> {
            future.await();
            executor.shutdownGracefully();
        });

        return promise;
    }

    /**
     * 发消息
     * <p>
     * 具体逻辑
     *
     * @param request 请求对象
     * @param clazz   接口class
     */
    private static void sendMsg0(Request request, Class<?> clazz) {
        String serviceName = getClassServerCache().get(clazz.getName());
        if (!StringUtils.hasText(serviceName)) {
            // 未找到接口与服务对应的关系，可能是未订阅、没有这个api，无论何种原因重新订阅一次试试看
            channelComponent.init();
            serviceName = getClassServerCache().get(clazz.getName());
        }
        if (!StringUtils.hasText(serviceName)) {
            PromiseUtil.get(request.getId()).setSuccess(Response.fail(request, "调用失败：未找到订阅服务..."));
            return;
        }
        Channel channel = channelComponent.getChannel(serviceName);
        if (Objects.isNull(channel)) {
            PromiseUtil.get(request.getId()).setSuccess(Response.fail(request, "调用失败：连接服务端失败..."));
            return;
        }
        channel.writeAndFlush(request).addListener(future -> {
            if (future.isSuccess()) {
                return;
            }
            if (future.cause() instanceof ClosedChannelException) {
                sendMsg0(request, clazz);
                return;
            }
            future.cause().printStackTrace();
            PromiseUtil.get(request.getId()).setSuccess(Response.fail(request, future.cause().getMessage()));
        });
    }
}
